package cn.dglydrpy.study.j2ee.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
/**
 * 拦截器 实现时需自行保证线程安全
 * @author LY_飘涯
 *
 */
public class ProducerInterceptorOne implements ProducerInterceptor<Integer, String> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerInterceptorOne.class);

    /**
     * 该方法封装进KafkaProducer.send()中，即运行在用户主线程中。Producer
确保在消息被序列化以计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修
改消息所属的topic和分区，否则会影响目标分区的计算。
     */
    @Override
    public ProducerRecord<Integer, String> onSend(ProducerRecord<Integer, String> record) {
        System.out.println("Producer拦截器1 -- go");


        // 消息发送的时候，经过拦截器，调用该方法

        // 要发送的消息内容
        final String topic = record.topic();
        final Integer partition = record.partition();
        final Integer key = record.key();
        final String value = record.value();
        final Long timestamp = record.timestamp();
        final Headers headers = record.headers();


        // 拦截器拦下来之后根据原来消息创建的新的消息
        ProducerRecord<Integer, String> newRecord = new ProducerRecord<Integer, String>(
                topic,
                partition,
                timestamp,
                key,
                "( " + value + ")", //改动消息
                headers
        );
        // 传递新的消息
        return newRecord;
    }

    /**
     * 该方法会在消息被应答之前或消息发送失败时调用，
并且通常都是在Producer回调逻辑触发之前。onAcknowledgement运行在Producer的IO线程中，因此不
要在该方法中放入很重的逻辑，否则会拖慢Producer的消息发送效率。
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("Producer1 -- back");
        // 消息确认或异常的时候，调用该方法，该方法中不应实现较重的任务
        // 会影响kafka生产者的性能。
    }

    //关闭Interceptor，主要用于执行一些资源清理工作
    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
    	
    	// 可获取producer传过来的自定义配置
    	
        final Object classContent = configs.get("classContent");
        System.out.println(classContent);
    }
}
